package com.shujia.flink.source;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Demo04MySource {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * env有两种添加Source的方式：
         * addSource：用于自定义Source的注册
         * fromSource：用于Flink自带的Source的注册
         */
        DataStreamSource<Integer> myDS = env.addSource(new MyDefineSource(2000));

        myDS.print();

        env.execute();

    }
}

class MyDefineSource implements SourceFunction<Integer> {
    Integer sleepTime;

    public MyDefineSource(int sleepTime) {
        this.sleepTime = sleepTime;
    }

    // run方法只会在程序启动时运行一次，运行完成后意味着程序结束
    @Override
    public void run(SourceContext<Integer> sourceContext) throws Exception {
        // 通过sourceContext将数据发送到下游
        while (true) {
            sourceContext.collect(1);
            Thread.sleep(this.sleepTime);
        }
    }

    // 程序结束时会执行
    @Override
    public void cancel() {

    }
}